package com.test;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.test.Utils.CsvReader;
import com.test.entity.StockTrade;
import com.test.service.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
public class Main {
    private static final String TOPIC = "stock-trades-";
    private static final String KAFKA_SERVERS = "192.168.88.135:19092,192.168.88.135:29092,192.168.88.135:39092";

    public static void main(String[] args) {
        try {
            System.out.println("Starting application...");
            
            // 发送数据到Kafka
            sendDataToKafka();
            System.out.println("Data sent to Kafka successfully");
            
            // 配置Flink执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 设置全局并行度为1
            env.setParallelism(1);
            // 配置重启策略
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));
            // 启用检查点
            env.enableCheckpointing(60000);
            
            // 设置水印生成间隔
            env.getConfig().setAutoWatermarkInterval(1000); // 每秒生成一次水印
            
            // 执行所有Flink作业
            executeFlinkJobs(env);
            
        } catch (Exception e) {
            log.error("Application error", e);
        }
    }

    private static void executeFlinkJobs(StreamExecutionEnvironment env) throws Exception {
        System.out.println("开始配置Flink作业...");
        
        // 在同一个环境中设置所有的数据流处理
        StockSellAndBuy.createDataStream(env);
        StockTradeAnalysis.createDataStream(env);
        TopTenPlatform.createDataStream(env);
        TopTenStock.createDataStream(env);
        TopTradePlace.createDataStream(env);
        TradeCal.createDataStream(env);

        System.out.println("所有作业配置完成，开始执行...");
        // 执行所有作业
        env.execute("Stock Analysis Jobs");
    }

    private static void sendDataToKafka() throws IOException {
        System.out.println("开始准备发送数据到Kafka...");
        // Kafka配置保持不变...
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
        System.out.println("Starting to send data to Kafka...");

        try (InputStream inputStream = Main.class.getClassLoader().getResourceAsStream("data/1.csv")) {
            if (inputStream == null) {
                throw new IOException("无法找到 resources 目录下的 CSV 文件");
            }

            List<StockTrade> trades = CsvReader.readCsv(inputStream);
            if (trades == null) {
                throw new IOException("读取 CSV 文件失败，无法获取数据");
            }
            System.out.println("Read " + trades.size() + " trades from CSV file.");
            int count = 0;

            for (StockTrade trade : trades) {
                try {
                    String message = objectMapper.writeValueAsString(trade);
                    kafkaTemplate.send(TOPIC, message);
                    if(count<=20){
                        System.out.println("成功发送消息到Kafka: " + message);
                            count++;
                    }

                } catch (Exception e) {
                    System.out.println("发送消息到Kafka失败"+ e);
                }
            }
        }
        System.out.println("所有数据发送完成");
    }
}